package com.dd.edu.rabbitmqboot.woker;


import com.rabbitmq.client.*;

import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.security.KeyManagementException;
import java.security.NoSuchAlgorithmException;
import java.util.concurrent.TimeoutException;

/**
 * RabbitMQ 是一个消息中间件Broker：接收消息，传递给下游应用
 * 术语：
 *  Producing 就是发送消息， 发送消息的程序是Producer
 *  Queue是RabbitMQ内部的一个组件，消息存储与Queue中。queue使用主机的内存和磁盘存储，受到内存和磁盘的限制
 *      可以想象为一个大的消息缓冲。很多Producer可以向同一个queue发送消息，很多消费者可以从 同一个queue消费消息
 *  Consuming就是接收消息。一个等待消费消息的程序称为Consumer
 *  生产者、消费者、队列不必在同一台机器，一般都是在不同的主机上的应用。
 *      一个应用可以同时是生产者和消费者
 */
public class HelloWorldSender {

    private static String queue_name = "hello";

    public static void main(String[] args) throws NoSuchAlgorithmException, KeyManagementException, URISyntaxException, InterruptedException {
        ConnectionFactory connectionFactory = new ConnectionFactory();
//        connectionFactory.setHost("192.168.233.102");
//        connectionFactory.setVirtualHost("/");
//        connectionFactory.setUsername("root");
//        connectionFactory.setPassword("123456");
//        connectionFactory.setPort(5672);
        // 协议： amqp:// 用户名：root 密码：123456 主机： 端口： 虚拟主机 /
        connectionFactory.setUri(URI.create("amqp://root:123456@192.168.233.102:5672/%2f"));

        // 消息发送
        messageSendOfQueue(connectionFactory);
//        messageSendOfExchange(connectionFactory);
//        messageConsumerOfExchangeLQ(connectionFactory);// 主动拉取
//        messageConsumerOfExchange(connectionFactory); // 被动接收

//        Thread.sleep(1000L);
//        // 消息接收
        messageReceiveOfQueue(connectionFactory);

    }

     private static void messageReceiveOfQueue(ConnectionFactory connectionFactory) {
        try{
            Connection connection = connectionFactory.newConnection();
            final Channel channel = connection.createChannel();

            channel.queueDeclare(queue_name, false, false, false, null);

            // 参数说明：队列名称、autoAck = true 是否自动确认 、 消息过来的回调函数、客户端忽略消息的回调函数
            channel.basicConsume(queue_name, true, (consumerTag, delivery) -> {
                final String message = new String(delivery.getBody(), "UTF-8");
                System.out.println("收到的消息="+ message);
            }, s -> {});

            Thread.sleep(1000L);
            System.out.println("消息接收完毕");

            // 手动确认
            channel.basicAck(1L, true);
//            channel.close();
            connection.close();

        } catch (TimeoutException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

     }

    private static void messageSendOfQueue(ConnectionFactory connectionFactory) {
        try(Connection connection = connectionFactory.newConnection();
            final Channel channel = connection.createChannel()){

            // 声明一个队列
            // 参数：队列名称、是否持久化、是否是排外队列（true客户端断开，队列自动删除）、是否自动删除（true队列不使用了自动删除）、队列属性map
            channel.queueDeclare(queue_name, false, false, false, null);

            final String message = "Hello World!";
            channel.basicPublish("", queue_name, null, message.getBytes());
            System.out.println(" send message = " + message);
        } catch (Exception e){
            e.printStackTrace();
        }
    }

    private static void messageSendOfExchange(ConnectionFactory connectionFactory) {
        try(Connection connection = connectionFactory.newConnection();
            final Channel channel = connection.createChannel()){

            // 队列声明： 队列名称、是否持久化、是否排他、是否自动删除、属性参数
            channel.queueDeclare("queue.biz", false, false, false, null);
            // 交换器声明： 交换器名称、交换器类型、是否持久化、是否自动删除、属性参数
            channel.exchangeDeclare("ex.biz", BuiltinExchangeType.DIRECT, false, false, null);
            // 队列绑定： 队列名称、 交换器名称、路由键
            channel.queueBind("queue.biz", "ex.biz", "hello.world");
            // 发布消息： 交换器名称、 路由键、 发送属性、 消息内容
            channel.basicPublish("ex.biz", "hello.world", null, "Hello World 1".getBytes());

        } catch (Exception e){
            e.printStackTrace();
        }
    }

    private static void messageReceiveOfExchangeLQ(ConnectionFactory connectionFactory) {
        try(Connection connection = connectionFactory.newConnection();
            final Channel channel = connection.createChannel()){

            // 主动拉取消息： 从队列； 指定是否自动确认消息
            final GetResponse getResponse = channel.basicGet("queue.biz", true);
            final byte[] body = getResponse.getBody();
            System.out.println("收到的消息：" + new String(body));

//                需要注意的 basicAck 方法需要传递两个参数
//                deliveryTag（唯一标识 ID）：当一个消费者向 RabbitMQ 注册后，会建立起一个 Channel ，RabbitMQ 会用 basic.deliver 方法向消费者推送消息，这个方法携带了一个 delivery tag， 它代表了 RabbitMQ 向该 Channel 投递的这条消息的唯一标识 ID，是一个单调递增的正整数，delivery tag 的范围仅限于 Channel
//                multiple：为了减少网络流量，手动确认可以被批处理，当该参数为 true 时，则可以一次性确认 delivery_tag 小于等于传入值的所有消息
            channel.basicAck(getResponse.getEnvelope().getDeliveryTag(),false);
        } catch (Exception e){
            e.printStackTrace();
        }
    }

    private static void messageConsumerOfExchange(ConnectionFactory connectionFactory) {
        try{Connection connection = connectionFactory.newConnection();
            final Channel channel = connection.createChannel();

            channel.basicConsume("queue.biz", new DeliverCallback() {
                @Override
                public void handle(String messageTag, Delivery delivery) throws IOException {
                    System.out.println("接收到的消息=" + new String(delivery.getBody()));
                }
            }, new CancelCallback() {
                @Override
                public void handle(String messageTag) throws IOException {
                    //
                }
            });

        } catch (Exception e){
            e.printStackTrace();
        }
    }
}
